package pku;

//import net.jpountz.lz4.LZ4BlockInputStream;

import io.Snappy;
import pku.serializers.OldMessageSerializer;

import java.io.*;
import java.util.*;
import java.util.zip.DataFormatException;
import java.util.zip.Inflater;

public class MessageReader implements Iterator<DefaultMessage> {
    private final String storePath;

    // 1st-level all folders
    private ArrayList<File[]> folderFilesList;
    private ArrayList<String> folderNameList;
    private int folderIndex;

    // 2nd-level one folder
    private File[] files;
    private int fileIndex;

    // 3rd-level one file
//    private BufferedReader bufferedReader;
    private BufferedInputStream bufferedInputStream;
    //    private String tmpBinString;
    private byte[] tmpMsg;

    public MessageReader(String storePath) {
        this.storePath = storePath;
    }

    private void fetchNextFolder() {
        do {
            folderIndex++;
            files = folderFilesList.get(folderIndex);
            fileIndex = 0;
        } while (files.length == 0 && folderIndex < folderFilesList.size() - 1);
    }

    private void fetchNextFile() {
        try {
            if (files.length == 0) {
                bufferedInputStream = null;
                return;
            }
            FileInputStream fi = new FileInputStream(files[fileIndex]);
//            LZ4BlockInputStream zip = new LZ4BlockInputStream(fi);
            //InflaterInputStream zip = new InflaterInputStream(fi, new Inflater(true));
            bufferedInputStream = new BufferedInputStream(fi);
//            bufferedReader = new BufferedReader(new InputStreamReader(zip));
        } catch (IOException e) {
            e.printStackTrace();
        } catch (NullPointerException e) {
            System.out.print("%%%%" + fileIndex + "," + folderIndex + "%%%%\n");
        }
    }

    public void attachNames(String queueName, Collection<String> topicNameList) {
        // folder_list
        folderNameList = new ArrayList<>(topicNameList.size());
        folderNameList.addAll(topicNameList);
        Collections.sort(folderNameList);


        // 1st-level: all folders
        folderFilesList = new ArrayList<>(folderNameList.size());
        folderNameList.forEach((folderString) -> {
            Vector<File> temFiles = new Vector<>();
            for (int i = 0; i < 4; i++) {
                File f = new File(storePath + File.separator + folderString + "_Thread-" + i);
                if (f.exists()) temFiles.add(f);
            }
            File[] files = new File[temFiles.size()];
            temFiles.toArray(files);
            folderFilesList.add(files);
        });

        folderIndex = -1;

        // 2nd-level: one folder
        fetchNextFolder();

        // 3rd-level: one file
        fetchNextFile();
//            tmpBinString = bufferedReader == null ? null : bufferedReader.readLine();
        tmpMsg = this.readData();
    }

    private void nextAndUpdateIteratorStates() throws IOException {
//        tmpBinString = bufferedReader.readLine();
        tmpMsg = this.readData();
        // 3rd-level: reach end of file
        if (tmpMsg == null || tmpMsg.length == 0) {
            // 2nd-level: reach end of files in this folder
            fileIndex++;
            if (fileIndex >= files.length) {
                if (folderIndex < folderFilesList.size() - 1) fetchNextFolder();
                else {
                    tmpMsg = null;
                    return;
                }
            }
            fetchNextFile();
//            tmpBinString = bufferedReader == null ? null : bufferedReader.readLine();
            tmpMsg = this.readData();
        }
    }

    private byte[] readData() {
        try {
            if (bufferedInputStream == null || bufferedInputStream.available() <= 0) return null;
            byte[] msgLen = new byte[3];
            boolean isCompressed = false;
            int r1 = bufferedInputStream.read(msgLen, 0, 3);

            if ((msgLen[2] & 0x80) == 0x80) {
                isCompressed = true;
                msgLen[2] &= 0x7f;
            }

            int len = Utils.bytes2int3(msgLen);
            byte[] msgData = new byte[len];
            int r2 = bufferedInputStream.read(msgData, 0, len);
//            if (isCompressed) msgData = uncompress(msgData);
            if (isCompressed) msgData = Snappy.uncompress(msgData, 0, len);
            return msgData;
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public byte[] uncompress(byte[] input) throws DataFormatException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        Inflater decompressor = new Inflater();
        try {
            decompressor.setInput(input);
            final byte[] buf = new byte[2048];
            while (!decompressor.finished()) {
                int count = decompressor.inflate(buf);
                bos.write(buf, 0, count);
            }
        } finally {
            decompressor.end();
        }
        return bos.toByteArray();
    }

    @Override
    public boolean hasNext() {
        return folderIndex < folderFilesList.size();
    }

    @Override
    public DefaultMessage next() {
        if (tmpMsg == null)
            return null;
        DefaultMessage message = OldMessageSerializer.bytes2message(tmpMsg);
        message.putHeaders(MessageHeader.TOPIC, folderNameList.get(folderIndex));
        try {
            nextAndUpdateIteratorStates();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return message;
    }
}
